package com.tqz.rocketmq.demo;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

/**
 * <p>
 * 消息的消费者，从 {@link Producer} 里面消费
 * </p>
 *
 * @author tian
 * @since 2021/1/17 14:01
 */
public class Consumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Constant.CONSUMER_GROUP);
        // 设置 Name Server  地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER_ADDR);
        // 消息从哪里开始消费，这里从最开始的位置消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 订阅消息，
        // 第一个参数为topic的名称
        // 第二个参数为tags的表达式，要匹配哪些规则，这里比kafka更灵活，类型rabbitmq的路由键，
        // * 代表所有
        consumer.subscribe(Constant.TOPIC, "*");

        // 监听消息，该模式是并行处理的，可能会有多个线程进行处理
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            msgs.forEach(msg -> System.out.println(new String(msg.getBody())));
            // 确认接收到消息
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        // 启动
        consumer.start();
    }
}
